package com.practice.car.cardataapp.schelduer.spark.task.complaints;

import com.practice.car.cardataapp.schelduer.spark.Schedule;
import com.practice.car.cardataapp.schelduer.spark.analy.ComplaintsRunner;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.io.Serializable;
import java.sql.Connection;
import java.sql.DriverManager;
import static com.practice.car.cardataapp.schelduer.spark.Schedule.getConnection;

/**
 * testCar project
 *
 * @author 陈春幸
 * @date 2020/9/9 0009
 */
public class ComplaintsAnalysis implements Serializable, ComplaintsRunner {
    private Connection conn = null;
    @Override
    public void run() {

        SparkSession session = SparkSession
                                .builder()
                                .master(Schedule.sparkMaster)
                                .appName("SparkAnalyzer2")
                                .getOrCreate();

        String path = Schedule.complaintsHdfsUrl;
        Dataset<Row> ds = session.read().json(path);
        getPlaceComplaint(ds);
        session.close();
    }

    public  void getPlaceComplaint(Dataset<Row> ds){/*城市投诉排行*/

        JavaPairRDD PlaceComplaints = ds.javaRDD().mapToPair(row -> {
            String string = row.getAs("place");
            String placeString = "其他地区";

            if (string != null) {
                placeString = string;
            }
            return new Tuple2<String, Integer>(placeString, 1);
        }).reduceByKey((Function2<Integer, Integer, Integer>) (v1, v2) -> v1 + v2);

        PlaceComplaints.foreach((VoidFunction<Tuple2<String, Integer>>) t -> {
            String sql = "insert into car_place_complaint (place, amount) values(" +
                    "\"" + t._1 + "\",\"" + t._2 + "\")";
            Connection conn = getConnection();
            conn.createStatement().execute(sql);
            //System.out.println(t._1+":"+t._2);
        });
    }
    //getConnection原来的位置
}
/*
    private Connection getConnection() {
        if (conn != null) {
            return conn;
        }
        try {
            //数据库配置 ，需要修改！！！
            conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdatadb?" +
                    "useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC", "root", "root");
        } catch (Exception e) {
            e.printStackTrace();
        }
        return conn;
    }

 */
